package com.suning.framework.es;

import com.suning.framework.utils.EsUtils;
import com.suning.framework.utils.JsonUtils;
import com.suning.framework.utils.QueryStringJSONBuilder;
import net.sf.json.JSONArray;
import net.sf.json.JSONObject;
import net.sf.json.util.JSONUtils;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.MapUtils;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.client.Client;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.sort.SortOrder;
import org.springframework.context.support.ClassPathXmlApplicationContext;

import java.util.*;

/**
 * 〈一句话功能简述〉<br>
 * 〈功能详细描述〉
 *
 * @author 16080951(林健)
 * @Copyright (c) 2016 苏宁软件技术有限公司 版权所有
 * @see [相关类/方法]（可选）
 * @since [产品 /模块版本] （可选）
 */
public class EsTest {

    //分片数。
    private final static int SHARDS = 5;

    //分页时，每页取多少条（默认）
    private final static int PAGE_SIZE = 10;

    //当前页(默认1)
    private final static int CURRENT_PAGE = 1;

    //分页时，多久算超时（秒）
    private final static int SCROLL_TIMEOUT = 1000 * 60 * 2;

    //查询结果最大容量（取最多100000条数据 ）
    private final static long MAX_LIST_CAPACITY = 100000;

    static {
        new ClassPathXmlApplicationContext("config/application.xml");
    }

    /**
     * 查询所有结果（非分页）
     *
     * @param <T>    the type parameter
     * @param jsonId 查询esmap中的jsonId。
     * @param clazz  最终转换的对象。
     * @param params 需要替换的参数。
     * @return list 返回封装好的数据结构。
     * @throws Exception the exception
     */
    public <T> List<T> queryForList(String jsonId, Class<T> clazz, Map<String, String> params) throws Exception {
        List<T> result = new ArrayList<T>();

        //查询语句
        String queryJson = EsConfig.getEsQueryConfig(jsonId).getQueryString();
        //查询索引
        String index = EsConfig.getEsQueryConfig(jsonId).getIndex();
        //查询类型
        String type = EsConfig.getEsQueryConfig(jsonId).getType();

        //若变量不变空，则替换里面的值。否则直接执行。
        if (MapUtils.isNotEmpty(params)) {
            Iterator paramsIterator = params.entrySet().iterator();
            while (paramsIterator.hasNext()) {
                Map.Entry entry = (Map.Entry) paramsIterator.next();
                String keys = (String) entry.getKey();
                Object values = null;
                if (entry.getValue() instanceof Number) {
                    values = (Number) entry.getValue();
                } else {
                    values = (String) entry.getValue();
                }
                queryJson = queryJson.replaceAll("#" + keys + "#", "\"" + values + "\"");
            }
        }

        Client client = EsUtils.openConnection();
        QueryBuilder queryStringJSONBuilder = new QueryStringJSONBuilder(JSONObject.fromObject(queryJson));
        SearchResponse searchResponse =
                client.prepareSearch(index).setTypes(type)
                        .setSearchType(SearchType.DFS_QUERY_AND_FETCH)
                        .setQuery(queryStringJSONBuilder)
                        .setTimeout(TimeValue.timeValueMillis(SCROLL_TIMEOUT))
                        //.setFrom(0)
                        //.setSize(10/5) //返回条数 = 分片数 * size
                        .execute()
                        .actionGet();

        //总量
        //;long total = searchResponse.getHits().getTotalHits();

        //将数据放入List
        Iterator<SearchHit> iterator = searchResponse.getHits().iterator();

        //聚合时，取出值
        //((StringTerms)searchResponse.getAggregations().get("namegroup")).getBuckets().get(0).getDocCount()
        //((StringTerms)searchResponse.getAggregations().get("namegroup")).getBuckets().get(0).getKey()

        //将数据序列化成对象
        while (iterator.hasNext()) {
            SearchHit hit = iterator.next();
            String json = hit.getSourceAsString();
            String _id = hit.getId();
            String _type = hit.getType();
            long _version = hit.getVersion();
            String _index = hit.getIndex();
            if (JSONUtils.mayBeJSON(json)) {
                JSONObject jsonObject = JSONObject.fromObject(json);
                jsonObject.put("_id", _id);
                jsonObject.put("_type", _type);
                jsonObject.put("_version", _version);
                jsonObject.put("_index", _index);
                result.add((T) JSONObject.toBean(jsonObject, clazz));
            }

            if (result.size() > MAX_LIST_CAPACITY)
                break;
        }

        return result;
    }

    /**
     * 按分页查询
     *
     * @param <T>    the type parameter
     * @param jsonId 查询esmap中的jsonId。
     * @param clazz  最终转换的对象。
     * @param params 需要替换的参数。
     * @return list 返回封装好的数据结构。
     * @throws Exception the exception
     */
    public <T extends EsBasePojo> List<T> queryForPage(String jsonId, Class<T> clazz, Map<String, Object> params) {
        List<T> result = new ArrayList<T>();
        //取出客户端
        Client client = EsUtils.openConnection();
        try {
            //查询语句
            String queryJson = EsConfig.getEsQueryConfig(jsonId).getQueryString();
            //查询索引
            String index = EsConfig.getEsQueryConfig(jsonId).getIndex();
            //查询类型
            String type = EsConfig.getEsQueryConfig(jsonId).getType();

            if (MapUtils.isNotEmpty(params)) {
                Iterator paramsIterator = params.entrySet().iterator();
                while (paramsIterator.hasNext()) {
                    Map.Entry entry = (Map.Entry) paramsIterator.next();
                    Object values = null;
                    if (entry.getValue() instanceof Number) {
                        values = (Number) entry.getValue();
                    } else {
                        values = (String) entry.getValue();
                    }
                    String keys = (String) entry.getKey();

                    queryJson = queryJson.replaceAll("#" + keys + "#", "\"" + values + "\"");
                }
            }

            //取出每页多少条
            int pageSize = MapUtils.getIntValue(params, "pageSize", PAGE_SIZE);
            pageSize = pageSize <= SHARDS ? SHARDS : pageSize;
            //当前页
            int currentPage = MapUtils.getIntValue(params, "currentPage", CURRENT_PAGE);

            QueryBuilder queryStringJSONBuilder = new QueryStringJSONBuilder(JSONObject.fromObject(queryJson));

            //执行搜索
            SearchResponse searchResponse =
                    client.prepareSearch(index)
                            .setTypes(type)
                            .setSearchType(SearchType.DFS_QUERY_AND_FETCH)
                            .setQuery(queryStringJSONBuilder)
                            .addSort("id", SortOrder.ASC)
                            .setScroll(TimeValue.timeValueMillis(SCROLL_TIMEOUT))
                            .setSize(pageSize)
                            .get();
            //总量
            long total = searchResponse.getHits().getTotalHits();

            //共多少页
            long totalPage = (total % pageSize == 0) ? total / pageSize : (total / pageSize) + 1;

            //若实际的页数不够总页数，默认以总页放入。
            currentPage = (int) (currentPage >= totalPage ? totalPage : currentPage);

            int currentClcr = 1;

            do {
                //清空数据
                result = new ArrayList<T>();
                //通过循环读取数据，将结果以List返回。
                for (SearchHit hit : searchResponse.getHits().getHits()) {
                    String json = hit.getSourceAsString();
                    String _id = hit.getId();
                    String _type = hit.getType();
                    long _version = hit.getVersion();
                    String _index = hit.getIndex();
                    if (JSONUtils.mayBeJSON(json)) {
                        JSONObject jsonObject = JSONObject.fromObject(json);
                        jsonObject.put("_id", _id);
                        jsonObject.put("_type", _type);
                        jsonObject.put("_version", _version);
                        jsonObject.put("_index", _index);
                        result.add((T) JSONObject.toBean(jsonObject, clazz));
                    }
                }
                //若==当前页。则返回结果。
                if(currentPage == currentClcr) return result;
                currentClcr++;

                //根据游标来计算下一趟取回的pageSize条数据。
                searchResponse = client.prepareSearchScroll(searchResponse.getScrollId())
                        .setScroll(TimeValue.timeValueMinutes(1))
                        .execute()
                        .actionGet();

            } while(searchResponse.getHits().getHits().length != 0);

            return result;

        } catch (Exception e) {
            e.printStackTrace();
            return result;
        }
    }

    /**
     * 根据条件删除测试
     *
     * @param <T>    the type parameter
     * @param jsonId 查询esmap中的jsonId。
     * @param clazz  最终转换的对象。
     * @param params 需要替换的参数。
     * @return list 返回封装好的数据结构。
     * @throws Exception the exception
     */
    public <T extends EsBasePojo> boolean deleteByQuery(String jsonId, Class<T> clazz, Map params) throws Exception {
        Client client = EsUtils.openConnection();
        try {
            while (true) {
                //查出匹配的结果
                List<T> result = queryForList(jsonId, clazz, params);
                if (CollectionUtils.isEmpty(result))
                    return true;

                BulkRequestBuilder bulkRequest = client.prepareBulk();
                for (T t : result) {
                    bulkRequest.add(
                            client.prepareDelete(t.get_index(), t.get_type(), t.get_id())
                    );
                }

                int count = bulkRequest.get().getItems().length;
                System.out.println("开始删除ES共 --->" + count + "条。");
                BulkResponse bulkResponse = bulkRequest.execute().actionGet();
                return bulkResponse.hasFailures();

            }
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 更新数据（_id,_type,_index)必传。
     *
     * @param <T>    the type parameter
     * @param updateData 需要更新的数据。
     * @return list 返回封装好的数据结构。
     * @throws Exception the exception
     */
    public <T extends EsBasePojo> boolean update(List<T> updateData) throws Exception {
        //取出客户端
        Client client = EsUtils.openConnection();
        try {
            if(CollectionUtils.isEmpty(updateData)) return false;
            BulkRequestBuilder bulkRequest = client.prepareBulk();
            for (T t : updateData) {

                JSONObject doc = JsonUtils.excloudsCustomProperty(t,new String[]{"_id","_index","_type","_version"});
                bulkRequest.add(
                client.prepareUpdate(t.get_index(), t.get_type(), t.get_id())
                        .setDoc(doc)
                );
            }

            int count = bulkRequest.get().getItems().length;
            System.out.println("开始更新ES共 --->" + count + "条。");
            BulkResponse bulkResponse = bulkRequest.execute().actionGet();
            return bulkResponse.hasFailures();

        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }


    /**
     * 更新数据（_id,_type,_index)必传。
     *
     * @param <T>    the type parameter
     * @param insertData 需要插入的数据。
     * @return list 返回封装好的数据结构。
     * @throws Exception the exception
     */
    public <T extends EsBasePojo> boolean insert(List<T> insertData) throws Exception {
        //取出客户端
        try {
            if(CollectionUtils.isEmpty(insertData)) return false;

            Client client = EsUtils.openConnection();
            BulkRequestBuilder bulkRequest = client.prepareBulk();

            for (T t : insertData) {
                //排除json指定属性。
                JSONObject doc = JsonUtils.excloudsCustomProperty(t,new String[]{"_id","_index","_type","_version"});
                bulkRequest.add(
                        client.prepareIndex(t.get_index(), t.get_type(), t.get_id())
                                .setSource(JsonUtils.formatJson(doc).toString())
                );
            }

            int count = bulkRequest.get().getItems().length;
            System.out.println("开始存入ES共 --->" + count + "条。");
            BulkResponse bulkResponse = bulkRequest.execute().actionGet();
            return bulkResponse.hasFailures();

        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 查询测试
     *
     * @throws Exception the exception
     */
    public static void queryTest() throws Exception {

        EsTest test = new EsTest();

        System.out.println("---------------开始聚合查询------------------------");
        List<QueryUserPojo> queryUserPojoListJuHe = test.queryForList("user.queryUserByAggs", QueryUserPojo.class,null);
        System.out.println(JSONArray.fromObject(queryUserPojoListJuHe,JsonUtils.formatJsonDate("yyyy-MM-dd")));

        //无参数查询
        System.out.println("---------------开始无参数查询1------------------------");
        List<QueryUserPojo> queryUserPojoList1 = test.queryForList("user.queryUser", QueryUserPojo.class,null);
        System.out.println(JSONArray.fromObject(queryUserPojoList1,JsonUtils.formatJsonDate("yyyy-MM-dd")));

        System.out.println("---------------开始无参数查询2------------------------");
        List<QueryUserPojo> queryUserPojoList2 = test.queryForList("user.queryUserByFilter", QueryUserPojo.class,null);
        System.out.println(JSONArray.fromObject(queryUserPojoList2,JsonUtils.formatJsonDate("yyyy-MM-dd")));

        System.out.println("---------------开始无参数查询3------------------------");
        List<QueryUserPojo> queryUserPojoList3 = test.queryForList("user.queryUserByConstant", QueryUserPojo.class,null);
        System.out.println(JSONArray.fromObject(queryUserPojoList3,JsonUtils.formatJsonDate("yyyy-MM-dd")));


        //可传参数查询
        System.out.println("---------------开始传Map查询------------------------");
        Map params = new HashMap();
        params.put("name","杨园园");
        params.put("age","25");
        params.put("update","2016-11-01");
        List<QueryUserPojo> queryMapUserPojoList3 = test.queryForList("user.queryUserByParams", QueryUserPojo.class,params);
        System.out.println(JSONArray.fromObject(queryMapUserPojoList3,JsonUtils.formatJsonDate("yyyy-MM-dd")));
        
        
        //分页查询
        System.out.println("---------------开始分页查询1------------------------");

        //可传参数查询
        params = new HashMap();
        params.put("pageSize",5);
        params.put("currentPage",1);
        List<QueryUserPojo> queryUserPojoPageList4 = test.queryForPage("user.queryUserByPage", QueryUserPojo.class,params);
        System.out.println(JSONArray.fromObject(queryUserPojoPageList4,JsonUtils.formatJsonDate("yyyy-MM-dd")));

        System.out.println("---------------开始分页查询2------------------------");
        params.put("pageSize",5);
        params.put("currentPage",2);
        queryUserPojoPageList4 = test.queryForPage("user.queryUserByPage", QueryUserPojo.class,params);
        System.out.println(JSONArray.fromObject(queryUserPojoPageList4,JsonUtils.formatJsonDate("yyyy-MM-dd")));
    }

    /**
     * 删除测试
     *
     * @throws Exception the exception
     */
    public static void deleteTest() throws Exception {
        EsTest test = new EsTest();
        System.out.println("---------------开始删除------------------------");
        //删除测试
        test.deleteByQuery("delete.user",QueryUserPojo.class,null);
    }

    /**
     * 更新测试
     *
     * @throws Exception the exception
     */
    public static void updateTest() throws Exception{
        System.out.println("---------------开始更新------------------------");
        EsTest test = new EsTest();
        List<QueryUserPojo> list = new ArrayList();
        QueryUserPojo q1 = new QueryUserPojo();
        q1.set_id("1");
        q1.set_index("mynote");
        q1.set_type("test");
        q1.setAge(30);
        q1.setName_keyword("张旋旋");
        q1.setName_text("张旋旋");
        list.add(q1);
        test.update(list);
    }

    /**
     * 更新测试
     *
     * @throws Exception the exception
     */
    public static void insertTest() throws Exception{
        System.out.println("---------------开始添加------------------------");
        EsTest test = new EsTest();
        List<QueryUserPojo> list = new ArrayList();
        initInsert(list);
        test.insert(list);
    }

    /**
     * 初始化插入数据
     * @param list
     */
    private static void initInsert(List<QueryUserPojo> list) {

        String[] name = { "杨园园","全欣","张小军","蔡冲冲","李国海","徐志欣","王桂斌","林健","叶庆国","史进","诗晶晶","吕德亮" };
        long[] age = {25,28,31,28,29,31,28,30,28,28,27,31};
        for(int i = 0;i<name.length;i++) {
            QueryUserPojo q1 = new QueryUserPojo();
            q1.set_id(String.valueOf(i));
            q1.setId(Long.valueOf(i));
            q1.set_index("mynote");
            q1.set_type("test");
            q1.setAge(age[i]);
            q1.setName_keyword(name[i]);
            q1.setName_text(name[i]);
            q1.setUpdate(new Date());
            list.add(q1);
        }

    }

    /**
     * The entry point of application.
     *
     * @param args the input arguments
     * @throws Exception the exception
     */
    public static void main(String[] args) throws Exception {
        //查询测试
        queryTest();
        //删除测试
        //deleteTest();
        //更新测试
        updateTest();
        //添加测试
        insertTest();
        //关闭
        EsUtils.closeConnection();

    }
}
